//
//  RACSignal+SYAdd.m
//  ReactiveCocoaTest
//
//  Created by huchu on 2018/5/15.
//  Copyright © 2018年 ff. All rights reserved.
//

#import "RACSignal+SYAdd.h"
#import "ReactiveObjC.h"

static RACDisposable *subscribeForever1 (RACSignal *signal, void (^next)(id), void (^error)(NSError *, RACDisposable *), void (^completed)(RACDisposable *) ,NSTimeInterval interval ) {
    next = [next copy];
    error = [error copy];
    completed = [completed copy];
    
    RACCompoundDisposable *compoundDisposable = [RACCompoundDisposable compoundDisposable];
    
    RACSchedulerRecursiveBlock recursiveBlock = ^(void (^recurse)(void)) {
        RACCompoundDisposable *selfDisposable = [RACCompoundDisposable compoundDisposable];
        [compoundDisposable addDisposable:selfDisposable];
        
        __weak RACDisposable *weakSelfDisposable = selfDisposable;
        
         RACDisposable *subscriptionDisposable = [signal subscribeNext:next error:^(NSError *e) {
             @autoreleasepool {
                 error(e, compoundDisposable);
                 [compoundDisposable removeDisposable:weakSelfDisposable];
             }
             RACScheduler *recursiveScheduler = RACScheduler.currentScheduler ?: [RACScheduler scheduler];
             [recursiveScheduler  afterDelay:interval schedule:^{
                 recurse();
             }];
             
         } completed:^{
             @autoreleasepool {
                 completed(compoundDisposable);
                 [compoundDisposable removeDisposable:weakSelfDisposable];
             }
             RACScheduler *recursiveScheduler = RACScheduler.currentScheduler ?: [RACScheduler scheduler];
             [recursiveScheduler  afterDelay:interval schedule:^{
                 recurse();
             }];
             
             
        }];
        
        [selfDisposable addDisposable:subscriptionDisposable];
    };
    
    // Subscribe once immediately, and then use recursive scheduling for any
    // further resubscriptions.
    recursiveBlock(^{
        RACScheduler *recursiveScheduler = RACScheduler.currentScheduler ?: [RACScheduler scheduler];
        
        RACDisposable *schedulingDisposable = [recursiveScheduler scheduleRecursiveBlock:recursiveBlock];
        [compoundDisposable addDisposable:schedulingDisposable];
    });
    
    return compoundDisposable;
}
@implementation RACSignal (SYAdd)

- (RACSignal *)retry:(NSInteger)retryCount interval :(NSTimeInterval)interval {
    return [[RACSignal createSignal:^(id<RACSubscriber> subscriber) {
        __block NSInteger currentRetryCount = 0;
         return subscribeForever1(self,
                                ^(id x) {
                                    [subscriber sendNext:x];
                                },
                                ^(NSError *error, RACDisposable *disposable) {
                                    if (retryCount == 0 || currentRetryCount < retryCount) {
                                        // Resubscribe.
                                        currentRetryCount++;
                                        return;
                                    }
                                    
                                    [disposable dispose];
                                    [subscriber sendError:error];
                                },
                                ^(RACDisposable *disposable) {
                                    [disposable dispose];
                                    [subscriber sendCompleted];
                                },
                                  interval);
    }] setNameWithFormat:@"[%@] -retry: %lu", self.name, (unsigned long)retryCount];
}

@end
